#include "shop_waiter_map.h"

#include <string>
#include <boost/algorithm/string.hpp>

#include "base/logging.h"
#include "base/stop_watch.h"
#include "base/config_reader.h"
#include "database/scoped_connection.h"
#include "database/simple_db_pool.h"
#include "thread_manager.h"

#include "im/adapters/session_adapter.h"

namespace imserver {

using apache::thrift::concurrency::Runnable;
class ReloadShopWaiterTask : public Runnable  {
 public:
  ReloadShopWaiterTask(){}
  virtual void run() {
    ShopWaiterMap::Instance().ReloadShopWaiter();
    ConfigReader config("../conf/im_call_center_service.conf");
    GetTimerManager()->add(boost::shared_ptr<Runnable>(new ReloadShopWaiterTask()), 140 * 1000);
  }
};

class ReloadOnlineWaitersTask : public Runnable  {
 public:
  ReloadOnlineWaitersTask(){}
  virtual void run() {
    ShopWaiterMap::Instance().ReloadOnlineWaiters();
    ConfigReader config("../conf/im_call_center_service.conf");
    GetTimerManager()->add(boost::shared_ptr<Runnable>(new ReloadOnlineWaitersTask()), 3 * 1000);
  }
};


ShopWaiterMap::ShopWaiterMap() : db_conn_pool_(NULL) {
  srand(time(NULL));
  ConfigReader config("../conf/im_call_center_service.conf");

  db_conn_pool_ = new base::SimpleConnectionPool(config.Get("database", "name").c_str(),
                      config.Get("database", "server_list").c_str(),
                      config.Get("database", "user").c_str(),
                      config.Get("database", "password").c_str());
  GetTimerManager()->add(boost::shared_ptr<Runnable>(new ReloadShopWaiterTask()), 300 * 1000);
  GetTimerManager()->add(boost::shared_ptr<Runnable>(new ReloadOnlineWaitersTask()), 3 * 1000);
}

int ShopWaiterMap::MapSenderReceiver(int64_t sender, int64_t receiver, int64_t * mapped_sender,
    int64_t * mapped_receiver) {
  *mapped_sender = sender;
  *mapped_receiver = receiver;
  boost::mutex::scoped_lock lock(mapping_mutex_);

  {
    boost::shared_lock<boost::shared_mutex> rlock(shop_waiters_mutex_);
    std::map<int64_t, std::vector<int64_t> >::iterator sw_it = shop_waiters_.find(receiver);
    if (sw_it != shop_waiters_.end()) { // 消息是否发给shop
      int64_t shop = receiver;
      std::map<int64_t, std::map<int64_t, int64_t> >::iterator cwm_it = customer_waiter_mapping_.find(shop);
      if (cwm_it != customer_waiter_mapping_.end()) {
        std::map<int64_t, int64_t>::iterator wm_it = cwm_it->second.find(sender);
        if (wm_it == cwm_it->second.end()) {
          // 找一个waiter
          int64_t waiter = GetShopRandomWaiter(receiver);
          if (waiter > 0) {
            cwm_it->second[sender] = waiter;
            *mapped_receiver = waiter;
          }
        } else {
          *mapped_receiver = wm_it->second; // 映射到合适的waiter

          boost::shared_lock<boost::shared_mutex> rlock(online_waiters_mutex_);
          if (online_waiters_.find(*mapped_receiver) == online_waiters_.end()) { // 映射的waiter不在线，则尝试修正
            std::map<int64_t, std::vector<int64_t> >::const_iterator sow_it = shop_online_waiters_.find(shop);
            if (sow_it != shop_online_waiters_.end()) {
              if (sow_it ->second.size() > 0) {
                size_t pos = rand() % sow_it->second.size();
                int64_t offline_waiter = *mapped_receiver;
                *mapped_receiver = sow_it->second[pos];
                wm_it->second = *mapped_receiver;
                waiter_customer_mapping_[shop][*mapped_receiver] = wm_it->first;
                LOG_INFO("shop " << shop << " offline waiter " << offline_waiter << " unmapped with "
                         << wm_it->first << ", newly mapped waiter=" << *mapped_receiver);
              } else {
                LOG_INFO("shop " << shop << " offline waiter " << *mapped_receiver << " need not unmapped.");
              }
            }
          }
        }
      } else {
        int64_t waiter = GetShopRandomWaiter(receiver);
        if (waiter > 0) {
          customer_waiter_mapping_[shop][sender] = waiter;
          *mapped_receiver = waiter;
          waiter_customer_mapping_[shop][waiter] = sender;
        }
      }
      return 1;
    }
  }

  {
    boost::shared_lock<boost::shared_mutex> rlock(waiter_shops_mutex_);
    std::map<int64_t, std::vector<int64_t> >::iterator ws_it = waiter_shops_.find(sender);
    if (ws_it != waiter_shops_.end()) { // 消息来自waiter
      int64_t waiter = sender;
      std::map<int64_t, std::map<int64_t, int64_t> >::iterator wcm_it = waiter_customer_mapping_.find(waiter);
      if (wcm_it != waiter_customer_mapping_.end()) {
        std::map<int64_t, int64_t>::iterator cm_it = wcm_it->second.find(receiver);
        if (cm_it == wcm_it->second.end()) {
          // 找一个shop
          int64_t shop = GetWaiterRandomShop(waiter);
          if (shop > 0) {
            wcm_it->second[receiver] = shop;
            *mapped_sender = shop;
          }
          // 尚未对应起来
        } else {
          int64_t shop = cm_it->second;
          *mapped_sender = shop; // 以shop的身份展示给接收方
        }
      } else {
        int64_t shop = GetWaiterRandomShop(waiter);
        if (shop > 0) {
          waiter_customer_mapping_[shop][waiter] = receiver;
          *mapped_sender = shop;
          customer_waiter_mapping_[shop][receiver] = waiter;
        }
      }
      return 2;
    }
  }

  return 0;
}

int64_t ShopWaiterMap::GetShopRandomWaiter(int64_t shop) const {
  // 在线的waiter优先
  {
    boost::shared_lock<boost::shared_mutex> rlock(online_waiters_mutex_);
    std::map<int64_t, std::vector<int64_t> >::const_iterator it = shop_online_waiters_.find(shop);
    if (it != shop_online_waiters_.end()) {
      if (it->second.size() > 0) {
        size_t pos = rand() % it->second.size();
        return it->second[pos];
      }
    }
  }

  {
    boost::shared_lock<boost::shared_mutex> rlock(shop_waiters_mutex_);
    std::map<int64_t, std::vector<int64_t> >::const_iterator it = shop_waiters_.find(shop);
    if (it != shop_waiters_.end()) {
      if (it->second.size() > 0) {
        size_t pos = rand() % it->second.size();
        return it->second[pos];
      }
    }
  }
  return 0;
}

int ShopWaiterMap::ReloadOnlineWaiters() {
  std::map<int64_t, SessionType::type> online_map;
  std::vector<int64_t> all_waiters;
  {
    boost::shared_lock<boost::shared_mutex> rlock(waiter_shops_mutex_);
    std::map<int64_t, std::vector<int64_t> >::iterator it = waiter_shops_.begin();
    for(; it != waiter_shops_.end(); ++it) {
      all_waiters.push_back(it->first);
    }
  }

  // 得到所有在线waiter
  SessionAdapter::Instance().GetOnlineUsers(all_waiters, &online_map);
  LOG_DEBUG("total online waiters=" << online_map.size());

  // shop_waiters_ 只可能在timer线程中被修改，所以在该线程中, shop_waiters_读无需lock
  std::map<int64_t, std::vector<int64_t> > shop_online_waiters;
  std::map<int64_t, std::vector<int64_t> >::iterator it = shop_waiters_.begin();
  for(; it != shop_waiters_.end(); ++it) {
    shop_online_waiters.insert(std::make_pair(it->first, std::vector<int64_t>()));
    for(size_t i = 0; i < it->second.size(); ++i) {
      if (online_map.find(it->second[i]) != online_map.end()) {
        shop_online_waiters[it->first].push_back(it->second[i]);
        LOG_DEBUG("shop " << it->first << " waiter " << it->second[i] << " is online.");
      } else {
        LOG_DEBUG("shop " << it->first << " waiter " << it->second[i] << " is offline.");
      }
    }
    LOG_DEBUG("shop " << it->first << " online waiters " << shop_online_waiters[it->first].size());
  }
  
  {
    boost::unique_lock<boost::shared_mutex> wlock(online_waiters_mutex_);
    shop_online_waiters.swap(shop_online_waiters_);
    online_waiters_.swap(online_map);
  }

  return 0;
}

int64_t ShopWaiterMap::GetWaiterRandomShop(int64_t waiter) const {
  boost::shared_lock<boost::shared_mutex> rlock(waiter_shops_mutex_);
  std::map<int64_t, std::vector<int64_t> >::const_iterator it = waiter_shops_.find(waiter);
  if (it != waiter_shops_.end()) {
    if (it->second.size() > 0) {
      size_t pos = rand() % it->second.size();
      return it->second[pos];
    }
  }
  return 0;
}

int ShopWaiterMap::ReloadShopWaiter() {
  mysqlpp::ScopedConnection db(*db_conn_pool_);
  if (!db) {
    LOG_ERROR("ShopWaiterMap::ReloadShopWaiter failed to get a connection from database-pool.");
    return -1;
  }

  std::map<int64_t, std::vector<int64_t> > shop_waiters;
  std::map<int64_t, std::vector<int64_t> > waiter_shops;

  mysqlpp::Query query(db->query());
  query << "SELECT shop, waiter FROM shop_waiters";

  try {
    mysqlpp::UseQueryResult ret = query.use();
    if (ret) {
      while(mysqlpp::Row row = ret.fetch_row()) {
        // TODO: 优化性能
        int64_t shop = row["shop"];
        int64_t waiter = row["waiter"];

        shop_waiters[shop].push_back(waiter);
        waiter_shops[waiter].push_back(shop);

        LOG_DEBUG("ShopWaiterMap::ReloadShopWaiter shop=" << shop
                  << " waiter=" << waiter);
      }
    }
  } catch (mysqlpp::Exception& e) {
    LOG_ERROR("ShopWaiterMap::ReloadShopWaiter " << query.str() << " EXEC ERROR.(" << e.what() << ")");
    return -2;
  }
//int64_t test_waiters[] = {101, 201, 288297970, 382097274};
//for(size_t i = 0; i < sizeof(test_waiters)/sizeof(test_waiters[0]); ++i) {
//  shop_waiters[127646892].push_back(test_waiters[i]);
//  waiter_shops[test_waiters[i]].push_back(127646892);
//}

  LOG_INFO("ShopWaiterMap::ReloadShopWaiter ok, shop_waiters.size=" << shop_waiters.size()
           << " waiter_shops.size=" << waiter_shops.size());
  {
    boost::unique_lock<boost::shared_mutex> wlock(shop_waiters_mutex_);
    shop_waiters.swap(shop_waiters_);
  }

  {
    boost::unique_lock<boost::shared_mutex> wlock(waiter_shops_mutex_);
    waiter_shops.swap(waiter_shops_);
  }
  return 0;
}

}

